package com.alison.datastream.chapter4_join;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.concurrent.TimeUnit;

/**
 * @Author alison
 * @Date 2024/4/5 16:26
 * @Version 1.0
 * @Description
 */
public class F1_WindowJoinExample {

    // 窗口join
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, Integer>> input1 = env.addSource(new MySource(1));
        DataStreamSource<Tuple2<String, Integer>> input2 = env.addSource(new MySource(2));
        input1.print("input1");
        input2.print("input2");

        input1.join(input2)
                .where(i1 -> i1.f0)
                .equalTo(i2 -> i2.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                // IN1, IN2, OUT
                .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String>() {
                    @Override
                    public String join(Tuple2<String, Integer> first, Tuple2<String, Integer> second) throws Exception {
                        return "input1: " + first.f1 + ", input2: " + second.f1;
                    }
                })
                .print();

        env.execute("window join function");
    }

    public static class MySource implements SourceFunction<Tuple2<String, Integer>> {
        private volatile boolean isRunning = true;

        int count = 1;

        public MySource(int count) {
            this.count = count;
        }

        @Override
        public void run(SourceContext ctx) throws Exception {

            while (isRunning) {
                Tuple2<String, Integer> data = Tuple2.of(String.valueOf(count), count);
                ctx.collect(data);
                count++;
                TimeUnit.SECONDS.sleep(2);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
